/* file name  : JSlingProvider.java
 * authors    : Saikat Guha <saikat@cs.cornell.edu>
 * created    : Fri 23 Sep 2005 10:26:12 PM EDT
 * copyright  : GNU Lesser General Public License version 2.1 or later
 * 
 * Copyright (C) 2005 Saikat Guha <saikat@cs.cornell.edu>
 * 
 * This library is free software; you can redistribute it and/or modify it under the terms
 * of the GNU Lesser General Public License as published by the Free Software Foundation;
 * either version 2.1 of the License, or (at your option) any later version.
 * 
 * This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
 *  without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * See the GNU Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public License along with this
 * library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330,
 * Boston, MA 02111-1307 USA
 * 
 */
package net.nutss.signaling;

import java.net.*;
import java.io.*;
import net.nutss.*;
import net.nutss.stunt.STUNTUDPSocket;

import static net.nutss.signaling.JSlingMessage.Type.*;
import static net.nutss.signaling.JSlingCommon.*;
import static net.nutss.Control.log;

/**
 * Stub signaling client for the JSling proxy server
 * 
 * @author Saikat Guha
 * @version %I%, %G%
 */
public class JSlingProvider implements SignalingContext {

	/**
	 * UDP socket to use for communication
	 */
	private STUNTUDPSocket sock;

	/**
	 * Address of proxy
	 */
	private SocketAddress addr;

	/**
	 * Receive buffer
	 */
	private byte[] bbuf =null;

	/**
	 * Implements a Keepalive protocol
	 * 
	 * @author Saikat Guha
	 * @version %I%, %G%
	 */
	private static class JSlingKeepalive implements Continuation {

		/**
		 * Retry timer for retransmitting lost packets.
		 */
		private static final int T_RETRY = 10000;

		/**
		 * Refresh timer for keepalive.
		 */
		//by casper
		private static final int T_EXPIRE = 40000;
		//private static final int T_EXPIRE = 59000;

		/**
		 * States of the Keepalive protocol
		 */
		enum State {
			/** Start state */
			START(0),

			/** Keepalive message sent */
			HELLO_SENT(T_RETRY),

			/** Keepalive message acknowledged */
			RESPONSE_RCVD(T_EXPIRE),

			/** Stop */
			STOP(-1);

			/**
			 * Timeout for the associated state in milliseconds
			 */
			private final int timeout;

			/**
			 * Creates a state with the specified timeout.
			 * 
			 * @param timeout
			 *            timeout in milliseconds
			 */
			State(int timeout) {
				this.timeout = timeout;
			}

			/**
			 * Returns the timeout of the state
			 * 
			 * @return timeout in milliseconds
			 */
			int getTimeout() {
				return timeout;
			}
		};

		/**
		 * Current state of the protocol
		 */
		private State state;

		/**
		 * Proxy context
		 */
		private JSlingProvider sig;

		/**
		 * Number of times the last message was retransmitted
		 */
		private int tries;

		/**
		 * Error to be reported to user
		 */
		private Exception reason = null;

		/**
		 * The ID to maintain the keepalive for
		 */
		private URI id;

		/**
		 * Keepalive packet
		 */
		private JSlingMessage pkt;

		/**
		 * Creates a new continuation for ID and socket
		 * 
		 * @param id
		 *            ID to register
		 * @param context
		 *            signaling channel
		 */
		JSlingKeepalive(URI id, JSlingProvider context)
				throws UnknownHostException, IOException {
			this.id = id;
			this.sig = context;
		}

		public SignalingMessage init(SignalingMessage msg) {
			try {
				// Registration message
				pkt = new JSlingMessage(id, new URI("registrar@nutss.net"),
						null);
				pkt.type = JSLING_REGISTER;
				state = State.START;
				return pkt;
			} catch (URISyntaxException e) {
				log.throwing("JSlingKeepalive", "init", e);
				return null;
			}
		}

		public int step(Object o, Dispatch dispatch,STUNTSelector sel) {
			try {
				log.entering("JSlingKeepalive", "step", o);
				if (o instanceof JSlingMessage) {
					JSlingMessage msg = (JSlingMessage) o;
					if (msg.isError()) {
						state = State.STOP;
						reason = new BindException("ID in use.");
					} else {
						state = State.RESPONSE_RCVD;
						if (msg.hasPayload())
							dispatch.callback(this, msg.getPayload());
					}
				} else {
					switch (state) {
					case START: // fallthrough
					case RESPONSE_RCVD:
						sig.addr = resolveProxy(id);
						sig.send(pkt);
						tries = 1;
						state = State.HELLO_SENT;
						break;

					case HELLO_SENT:
						if (tries++ < 3) {
							sig.send(pkt);
						} else {
							state = State.STOP;
							reason = new SocketTimeoutException(
									"Binding Timeout");
						}
						break;
					}
				}
			} catch (IOException e) {
				log.throwing("JSlingKeepalive", "step", e);
				reason = e;
				state = State.STOP;
			}
			return state.getTimeout();
		}

		public void cancel(Dispatch d) {
			if (reason != null) {
				log.throwing("JSlingKeepalive", "cancel", reason);
				d.callback(this, reason);
			}
		}
	}

	/**
	 * Creates a new signaling connection
	 */
	public JSlingProvider() {
	}

	public SignalingMessage receive(byte[] bbuf) {
		try {
			// Use globally preallocated buffer
			ByteArrayInputStream bis = new ByteArrayInputStream(bbuf);
			ObjectInputStream ois = new ObjectInputStream(bis);
			SignalingMessage msg = (SignalingMessage) ois.readObject();
			ois.close();
			return msg;
		} catch (Exception e) {
			log.throwing("JSlingProvider", "run", e);
			return null;
		}
	}

	/**
	 * Check that the SignalingMessage is a JSlingMessage
	 * 
	 * @param msg
	 *            the message to check
	 * @return the message cast as a JSlingMessage
	 * @throws IllegalArgumentException
	 *             if the message is not a JSlingMessage
	 */
	private JSlingMessage check(SignalingMessage msg) {
		if (msg instanceof JSlingMessage) {
			return (JSlingMessage) msg;
		} else {
			throw new IllegalArgumentException("JSlingMessage required");
		}
	}

	public void send(SignalingMessage message) throws IOException {
		JSlingMessage msg = check(message);
		if (msg.bbuf == null) {
			// cache serialized version; re-used when resending the
			// same packet
			ByteArrayOutputStream bos = new ByteArrayOutputStream();
			ObjectOutputStream oos = new ObjectOutputStream(bos);
			oos.writeObject(msg);
			oos.close();
			msg.bbuf = bos.toByteArray();
		}
		sock.send(msg.bbuf,addr);
		log.fine("sending to " + addr + ": msg=" + msg);
	}

	public Continuation register(URI localAddress,STUNTSelector sel) throws UnknownHostException {
		try {
			addr = resolveProxy(localAddress);
			Continuation k=new JSlingKeepalive(localAddress, this);
			DatagramSocket sock1 = new DatagramSocket();
			this.sock= new STUNTUDPSocket(sock1,k,sel);
			this.sock.start();
			return k;
		} catch (IOException e) {
			log.throwing("JSlingProvider", "register", e);
			return null;
		}
	}

	public SignalingMessage createMessage(URI localAddress, URI remoteAddress,
			Object payload) {
		return new JSlingMessage(localAddress, remoteAddress, payload);
	}

	public SignalingMessage createReply(SignalingMessage msg, Object payload) {
		JSlingMessage jmsg = check(msg);
		JSlingMessage rmsg = new JSlingMessage(jmsg.getDestination(), jmsg
				.getSource(), payload);
		rmsg.id = jmsg.id;
		return rmsg;
	}

	public SignalingMessage setMessage(SignalingMessage msg, Object payload) {
		JSlingMessage jmsg = check(msg);
		jmsg.payload = payload;
		jmsg.bbuf = null; // invalidate cached serialization
		return msg;
	}

	public void close() {
		try {
			sock.close();
		} catch (Exception e) {
			log.throwing("JSlingProvider", "close", e);
		}
	}

}
